package cn.linkai.hadoop.topmax3;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;
import java.util.*;

public class Top3Max {
    public static void main(String[] args) {
        Configuration conf=new Configuration();
        Job job=null;
        try {
            job= Job.getInstance(conf);
            job.setJarByClass(Top3Max.class);
            job.setJobName("top3Max");

            job.setMapperClass(Top3MaxMapper.class);
            job.setCombinerClass(Top3MaxCombiner.class);
            job.setReducerClass(Top3MaxReducer.class);

            job.setMapOutputKeyClass(Text.class);
            job.setMapOutputValueClass(LongWritable.class);
            job.setOutputKeyClass(LongWritable.class);
            job.setOutputValueClass(Text.class);

            FileInputFormat.addInputPath(job,new Path("E:\\hadoop\\hadoopdata\\20160430105759102\\maxdata"));
            FileOutputFormat.setOutputPath(job,new Path("E:\\hadoop\\hadoopdata\\out20160430105759102"));

            System.out.println(job.waitForCompletion(true)?"success":"fail");

        } catch (IOException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ClassNotFoundException e) {
            e.printStackTrace();
        }
    }

    public static class Top3MaxMapper extends Mapper<LongWritable,Text,Text,LongWritable>{

        private static final Text NUMFLAG=new Text("NUM");

        @Override
        protected void map(LongWritable key,Text value, Context context) throws IOException, InterruptedException {
            context.write(NUMFLAG,new LongWritable(Long.valueOf(value.toString())));
        }
    }

    public static class Top3MaxCombiner extends Reducer<Text,LongWritable,Text,LongWritable>{

        private static final Text TOPFLAG=new Text("TOP");

        @Override
        protected void reduce(Text key, Iterable<LongWritable> values, Context context) throws IOException, InterruptedException {
            Iterator<LongWritable> it=values.iterator();
            List<Long> numbers=new ArrayList<>();
            while(it.hasNext()){
                numbers.add(it.next().get());
            }

            Collections.sort(numbers, new MyComparator());
            for(int i=0;i<3;i++){
                context.write(TOPFLAG,new LongWritable(numbers.get(i)));
            }

        }
    }

    public static class Top3MaxReducer extends Reducer<Text,LongWritable,LongWritable,Text>{

        @Override
        protected void reduce(Text key, Iterable<LongWritable> values, Context context) throws IOException, InterruptedException {
            if(!key.toString().equals("TOP")) return;

            Iterator<LongWritable> it=values.iterator();
            List<Long> numbers=new ArrayList<>();
            while(it.hasNext()){
                numbers.add(it.next().get());

            }
            Collections.sort(numbers,new MyComparator());
            numbers=numbers.subList(0,3);
            for(Long e:numbers){
                context.write(new LongWritable(e),new Text(""));
            }
        }
    }

    static class MyComparator implements Comparator<Long>{

        @Override
        public int compare(Long o1, Long o2) {
            if(o1>o2){
                return -1;
            }else if(o1<o2){
                return 1;
            }else{
                return 0;
            }
        }
    }

}
